package org.mj.im.common.rocketmq;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.mj.im.common.MessageSenderService;
import org.mj.im.common.constants.IMConstants;
import org.mj.im.common.model.TopicMessage;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Objects;

@Component
@ConditionalOnProperty(value = "message.mq.type",havingValue = "rocketmq")
public class RocketMQMessageSenderService implements MessageSenderService {

    private final RocketMQTemplate rocketMQTemplate;

    public RocketMQMessageSenderService(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    @Override
    public boolean send(TopicMessage topicMessage) {
        try{
            SendResult sendResult = this.rocketMQTemplate.syncSend(topicMessage.getDestination(), this.getMessage(topicMessage));
            return Objects.equals(sendResult.getSendStatus(),SendStatus.SEND_OK);
        }catch (Exception e){
            return false;
        }
    }

    @Override
    public TransactionSendResult sendMessageInTransaction(TopicMessage topicMessage, Object arg) {
        return this.rocketMQTemplate.sendMessageInTransaction(topicMessage.getDestination(),this.getMessage(topicMessage),arg);
    }

    private Message<String> getMessage(TopicMessage topicMessage) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put(IMConstants.MSG_KEY,topicMessage);
        return MessageBuilder.withPayload(jsonObject.toJSONString()).build();
    }


}
